package com.io.example.demoparent.mq.work;


import com.io.example.demoparent.mq.util.RabbitMQUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 *  工作模式
 */
public class Task01 {

    public static final  String QUEUE_NAME="ack_queue";

    /*
     线程安全有序的一个哈希表，适用于高并发的情况下
     1.将序列号与消息进行关联
     2.将回调成功的数据删除
     3.支持高并发（多线程）
      */

    public static void main(String[] args) throws Exception{
        publishMessageAsync();
    }
    /*
         发布消息后确认-非异步
     */
    public static  void  publishMessage() throws Exception{
        Channel channel = RabbitMQUtil.getRabbitChannel( );

        /*
         *   生产一个队列
         *   1.队列名称
         *   2.队列里面的消息是否实持久化，默认情况储存在内存中
         *   3.该队列是否提供一个消费者，是否进行消息共享
         *   4.之是否自动删除
         *   5.其他
         */
        channel.queueDeclare(QUEUE_NAME,Boolean.TRUE,false,false,null);
        channel.confirmSelect();
        long begin = System.currentTimeMillis( );
        Scanner scanner=new Scanner(System.in);
        while (scanner.hasNext()) {
            /*
             * 发送一个消息
             * 1.发送到那个交换机
             * 2.路由的key值是那个
             * 3.消息是否持久化  MessageProperties.PERSISTENT_TEXT_PLAIN
             * 4.消息体
             *
             */
            String message = scanner.next( );
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
            long end = System.currentTimeMillis( );
            boolean waitForConfirms = channel.waitForConfirms( );
            if (waitForConfirms) {
                System.out.println("消息发送成功");
            }

            System.out.println("单独确定发布消息，耗时" + (end - begin) + "ms");
        }
    }

    /*
         发布消息后确认-异步
     */
    public static void publishMessageAsync() throws Exception {

        ConcurrentSkipListMap<String, Object> concurrentSkipListMap= new ConcurrentSkipListMap<String,Object>();

        Channel channel = RabbitMQUtil.getRabbitChannel( );
        /*
         *   生产一个队列
         *   1.队列名称
         *   2.队列里面的消息是否实持久化，默认情况储存在内存中
         *   3.该队列是否提供一个消费者，是否进行消息共享
         *   4.之是否自动删除
         *   5.其他
         */
        channel.queueDeclare(QUEUE_NAME,Boolean.TRUE,Boolean.FALSE,Boolean.FALSE,null);

        //开启发布确认
        channel.confirmSelect();
        long begin = System.currentTimeMillis( );
        //准备监听器 监听消息是否发送成功
        channel.addConfirmListener(new ConfirmCallback( ) {
            @Override
            public void handle(long l, boolean b) throws IOException {
                if (b){
                    ConcurrentNavigableMap<String, Object> confirmed = concurrentSkipListMap.headMap(l + "");
                    confirmed.clear();
                }else {
                    concurrentSkipListMap.remove(l+"");
                }
                //消息确定成功 回调
                System.out.println("成功----消息标识："+l+"是否是批量数据："+b );
            }
        }, new ConfirmCallback( ) {
            @Override
            public void handle(long l, boolean b) throws IOException {
                Object o = concurrentSkipListMap.get(l + "");

                //消息确定失败 回调
                System.out.println("消息没有成功"+l +"消息内容："+o);
            }
        });


     for (int i = 0; i <1000; i++ ) {
            /*
             * 发送一个消息
             * 1.发送到那个交换机
             * 2.路由的key值是那个
             * 3.消息是否持久化  MessageProperties.PERSISTENT_TEXT_PLAIN
             * 4.消息体
             *
             */
             String message="发送消息："+i;
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
            // 1.记录发送的
             concurrentSkipListMap.put(channel.getNextPublishSeqNo()+"",message);

     }
        long end = System.currentTimeMillis( );
        System.out.println("批量发送，耗时" + (end - begin) + "ms");


    }




}
